/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.naming.core.v2.index;

import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.common.notify.listener.SmartSubscriber;
import com.alibaba.nacos.common.utils.ConcurrentHashSet;
import com.alibaba.nacos.naming.core.v2.client.Client;
import com.alibaba.nacos.naming.core.v2.event.client.ClientEvent;
import com.alibaba.nacos.naming.core.v2.event.client.ClientOperationEvent;
import com.alibaba.nacos.naming.core.v2.event.publisher.NamingEventPublisherFactory;
import com.alibaba.nacos.naming.core.v2.event.service.ServiceEvent;
import com.alibaba.nacos.naming.core.v2.pojo.Service;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Client and service index manager.
 * 事件处理类
 *
 * @author xiweng.yy
 */
@Component
public class ClientServiceIndexesManager extends SmartSubscriber {

    /**
     * 服务端本地注册表
     * key：Service
     * value：Set<clientId>
     */
    private final ConcurrentMap<Service, Set<String>> publisherIndexes = new ConcurrentHashMap<>();

    /**
     * 服务订阅表，缓存某个服务的所有订阅者列表
     * key：服务
     * value：某个服务的所有订阅者集合
     */
    private final ConcurrentMap<Service, Set<String>> subscriberIndexes = new ConcurrentHashMap<>();
    
    public ClientServiceIndexesManager() {
        /**
         * 注册订阅者，将自己 ClientServiceIndexesManager 注册进了 NamingEventPublisher
         * NamingEventPublisher 监听到相关事件时，会向所有的订阅者发布事件，例如会调用 ClientServiceIndexesManager#onEvent()
         */
        NotifyCenter.registerSubscriber(this, NamingEventPublisherFactory.getInstance());
    }

    /**
     * 从服务端本地注册表中获取目标服务的所有 clientId 的 set 集合
     * @param service
     * @return
     */
    public Collection<String> getAllClientsRegisteredService(Service service) {
        /**
         * 从服务端本地注册表中获取目标服务的所有 clientId 的 set 集合
         * 若服务端本地注册表中没有数据，则返回一个空集合
         */
        return publisherIndexes.containsKey(service) ? publisherIndexes.get(service) : new ConcurrentHashSet<>();
    }

    /**
     * 从订阅表中获取目标服务的所有订阅者集合，若没有，则返回一个空集合
     * @param service
     * @return
     */
    public Collection<String> getAllClientsSubscribeService(Service service) {
        return subscriberIndexes.containsKey(service) ? subscriberIndexes.get(service) : new ConcurrentHashSet<>();
    }
    
    public Collection<Service> getSubscribedService() {
        return subscriberIndexes.keySet();
    }
    
    /**
     * Clear the service index without instances.
     *
     * @param service The service of the Nacos.
     */
    public void removePublisherIndexesByEmptyService(Service service) {
        if (publisherIndexes.containsKey(service) && publisherIndexes.get(service).isEmpty()) {
            publisherIndexes.remove(service);
        }
    }

    /**
     * 订阅处理的事件类型
     * @return
     */
    @Override
    public List<Class<? extends Event>> subscribeTypes() {
        List<Class<? extends Event>> result = new LinkedList<>();
        // 注册类型事件
        result.add(ClientOperationEvent.ClientRegisterServiceEvent.class);
        // 注销类型事件
        result.add(ClientOperationEvent.ClientDeregisterServiceEvent.class);
        // 订阅类型事件
        result.add(ClientOperationEvent.ClientSubscribeServiceEvent.class);
        // 取消订阅类型事件
        result.add(ClientOperationEvent.ClientUnsubscribeServiceEvent.class);
        // 断开连接类型事件
        result.add(ClientEvent.ClientDisconnectEvent.class);
        return result;
    }

    /**
     * 处理事件
     * @param event {@link Event}
     */
    @Override
    public void onEvent(Event event) {
        if (event instanceof ClientEvent.ClientDisconnectEvent) {
            // 处理客户端断开连接事件
            handleClientDisconnect((ClientEvent.ClientDisconnectEvent) event);
        } else if (event instanceof ClientOperationEvent) {
            // 处理其它类型事件(注册、注销、订阅、取消订阅类型的事件)
            handleClientOperation((ClientOperationEvent) event);
        }
    }
    
    private void handleClientDisconnect(ClientEvent.ClientDisconnectEvent event) {
        Client client = event.getClient();
        for (Service each : client.getAllSubscribeService()) {
            // 删除订阅表中的失效(断开)的连接
            removeSubscriberIndexes(each, client.getClientId());
        }
        for (Service each : client.getAllPublishedService()) {
            // 删除注册表中的失效(断开)的连接
            removePublisherIndexes(each, client.getClientId());
        }
    }

    /**
     * 处理事件(注册、注销、订阅、取消订阅类型的事件)
     * @param event
     */
    private void handleClientOperation(ClientOperationEvent event) {
        Service service = event.getService();
        String clientId = event.getClientId();
        if (event instanceof ClientOperationEvent.ClientRegisterServiceEvent) {
            // 处理客户端注册事件
            addPublisherIndexes(service, clientId);
        } else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) {
            // 处理客户端注销事件
            removePublisherIndexes(service, clientId);
        } else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) {
            /**
             * 处理客户端订阅事件
             * service：某个被订阅的服务
             * clientId：这个服务的订阅者
             */
            addSubscriberIndexes(service, clientId);
        } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) {
            // 处理客户端取消订阅事件
            removeSubscriberIndexes(service, clientId);
        }
    }
    
    private void addPublisherIndexes(Service service, String clientId) {

        /**
         * 向服务端注册表(publisherIndexes)中添加服务数据
         * key：Service
         * value：ConcurrentHashSet 集合，用于保存客户端的 ClientId
         */
        publisherIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>());

        /**
         * 客户端真正的注册
         * 将客户端的 ClientId 保存进 ConcurrentHashSet 集合中，也即保存进了服务端注册表 publisherIndexes 中了
         * publisherIndexes 为服务端注册表，key：service，value：set<clientId>
         */
        publisherIndexes.get(service).add(clientId);

        // 至此，客户端完成了注册，然后发布服务变更(完成注册)事件(ServiceChangedEvent)
        NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));
    }
    
    private void removePublisherIndexes(Service service, String clientId) {
        if (!publisherIndexes.containsKey(service)) {
            return;
        }
        publisherIndexes.get(service).remove(clientId);
        NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));
    }
    
    private void addSubscriberIndexes(Service service, String clientId) {
        // 服务订阅表，缓存订阅者，缓存某个服务的所有订阅者列表
        subscriberIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>());
        // Fix #5404, Only first time add need notify event.
        /**
         * 将订阅者添加进目标服务的订阅者集合中，即给目标服务曾加一个订阅者
         */
        if (subscriberIndexes.get(service).add(clientId)) {
            // 发布服务订阅完毕事件
            NotifyCenter.publishEvent(new ServiceEvent.ServiceSubscribedEvent(service, clientId));
        }
    }
    
    private void removeSubscriberIndexes(Service service, String clientId) {
        if (!subscriberIndexes.containsKey(service)) {
            // 订阅注册表中没有该连接，则直接跳过
            return;
        }

        // 在订阅者注册表中删除该服务对应的连接
        subscriberIndexes.get(service).remove(clientId);
        if (subscriberIndexes.get(service).isEmpty()) {
            // 若订阅注册表中当前服务没有任何连接，则说明当前服务没有任何实例注册到注册中心，因此将当前服务从订阅注册表中删除
            subscriberIndexes.remove(service);
        }
    }
}
